package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.heima.model.common.constants.HotArticleConstants;
import com.heima.model.message.ArticleVisitStreamMess;
import com.heima.model.message.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
@Slf4j
public class HandleStream {

    /**
     * 流式处理的业务
     * @param streamsBuilder
     * @return  聚合处理之后，我们希望的消息内容："articleId:12,views:1,collection:3,comment:4,likes:5"
     */
    @Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
        //1.接收消息
        KStream<String,String> kStream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        //2.聚合处理
        kStream.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            /**
             * 对接收的消息进行初始化操作
             * @param key = null
             * @param value=json字符串的updateArticleMess
             * @return
             *      返回之后的key定义为articleId,value=type:add
             */
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                /**
                 *  1.转对象
                 *  articleId
                 *  type
                 *  add
                 */
                UpdateArticleMess mess = JSONObject.parseObject(value, UpdateArticleMess.class);
                return new KeyValue<>(mess.getArticleId().toString(),mess.getType()+":"+mess.getAdd());
            }
        })
                /**
                 * 分组操作
                 * key=articleId
                 * value=type:add
                 */
                .groupBy((key,value)-> key)
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                //聚合处理
                .aggregate(new Initializer<String>() {
                    /**
                     *聚合初始化方法
                     * 想要形成什么样的效果 "articleId:12,views:1,collection:3,comment:4,likes:5"
                     */
                    @Override
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                },
                        /**
                         * 具体实现聚合处理的核心方法
                         */
                        new Aggregator<String, String, String>() {
                            /**
                             *  流式处理的业务
                             * @param key=articleId
                             * @param value=type:add
                             * @param init ="COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0"
                             * @return 业务完成之后，返回的字符串="COLLECTION:0,COMMENT:2,LIKES:0,VIEWS:0"
                             */
                    @Override
                    public String apply(String key, String value, String init) {
                        //1.判断value是否为空
                        if (StringUtils.isEmpty(value)) {
                            return init;
                        }

                        //定义变量
                        int collection = 0,comment=0,like=0,view=0;

                        //2.对value进行分割成数组
                        String[] split = value.split(":");
                        switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                            case LIKES:
                                like+=Integer.parseInt(split[1]);
                                break;
                            case VIEWS:
                                view+=Integer.parseInt(split[1]);
                                break;
                            case COMMENT:
                                comment+=Integer.parseInt(split[1]);
                                break;
                            case COLLECTION:
                                collection+=Integer.parseInt(split[1]);
                                break;
                        }
                        String format = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", collection, comment, like, view);
                        log.info("流式处理之后的结果:{}",format);
                        return format;
                    }
                },Materialized.as("hot-article-stream"))
                .toStream()
                /**
                 * 准备聚合流式处理之后的消息内容，准备发送消息
                 */
                .map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
                    /**
                     *
                     * @param key =articleId
                     * @param value="COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d"
                     * @return
                     */
                    @Override
                    public KeyValue<String, String> apply(Windowed<String> key, String value) {
                        return new KeyValue<>(key.key().toString(),format(key.key().toString(),value.toString()));
                    }
                })
                //3.发送消息
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);

        return  kStream;
    }

    /**
     * 封装发送消息的内容
     * @param key =articleId
     * @param value="COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d"
     * @return
     */
    private String format(String key,String value){
        ArticleVisitStreamMess mess=new ArticleVisitStreamMess();
        mess.setArticleId(Long.parseLong(key));
        //split[0]="COLLECTION:%d"
        String[] split = value.split(",");
        for (String agg : split) {
            String[] aggSplit = agg.split(":");
            switch (UpdateArticleMess.UpdateArticleType.valueOf(aggSplit[0])){
                case LIKES:
                    mess.setLike(Integer.parseInt(aggSplit[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(aggSplit[1]));
                    break;
                case COMMENT:
                   mess.setComment(Integer.parseInt(aggSplit[1]));
                    break;
                case COLLECTION:
                  mess.setCollect(Integer.parseInt(aggSplit[1]));
                    break;
            }
        }
        return JSON.toJSONString(mess);
    }
}
